package cn.me.alphamq.broker.store;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.config.BrokerProperties;
import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueConsumeReq;
import cn.me.alphamq.broker.model.timewheel.DelayMessageModel;
import cn.me.alphamq.broker.enumeration.SlotMsgTypeEnum;
import cn.me.alphamq.broker.core.timewheel.TimeWheelModelExecutor;
import cn.me.alphamq.broker.model.timewheel.TimeWheelSlotModel;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.PushDelayMsgReq;
import cn.me.alphamq.common.util.DelayTimeUtil;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * broker 故障重启后的数据恢复管理器
 *
 * @author f
 */
public class RecoverManager {

    private static final Logger log = LoggerFactory.getLogger(RecoverManager.class);

    /**
     * 恢复数据
     */
    public static void recover() {
        recoverDelayMessage();
        log.info("recover ok");
    }

    // 恢复延迟消息数据 TODO 如果没有这个队列，应该创建一个，可以在初始化完毕后使用 Check 类检查一下是否有这些必要的队列
    private static void recoverDelayMessage() {
        BrokerProperties brokerProperties = Container.getBrokerProperties();
        int currDelayQueueId = brokerProperties.getBrokerCurrDelayQueueId();
        int otherDelayQueueId = brokerProperties.getBrokerOtherDelayQueueId();
        ConsumeQueueConsumeReq consumeQueueConsumeReq = new ConsumeQueueConsumeReq(
                BrokerConstant.DELAY_TOPIC_NAME, BrokerConstant.RECOVER_DELAY_MSG_CONSUMER_GROUP,
                otherDelayQueueId, BrokerConstant.DELAY_MSG_PULL_NUM
        );
        while (true) {
            List<Message> messageList = Container.getConsumeQueueConsumeHandler().consume(consumeQueueConsumeReq);
            if (messageList == null || messageList.isEmpty()) {
                break;
            }

            messageList.forEach(message -> {
                PushDelayMsgReq pushDelayMsgReq = JSON.parseObject(message.getBody(), PushDelayMsgReq.class);
                long executeTime = pushDelayMsgReq.getExecuteTime();
                if (DelayTimeUtil.getDelayTime(executeTime) <= 0) {
                    // 如果没有剩余延迟时间，则直接执行
                    TimeWheelModelExecutor.execute(new TimeWheelSlotModel(
                            pushDelayMsgReq, SlotMsgTypeEnum.DELAY_MSG.getCode(), executeTime
                    ), false);
                } else {
                    // 如果有剩余延迟时间，则将其投放到时间轮中
                    TimeWheelModelManager.add(DelayMessageModel.newDelayMsgModel(pushDelayMsgReq));
                    // 将消息持久化到当前延迟队列中
                    try {
                        Container.getCommitLogAppendHandler().appendMsg(Message.newDelayMsg(
                                BrokerConstant.DELAY_TOPIC_NAME,
                                currDelayQueueId,
                                JSON.toJSONBytes(pushDelayMsgReq)
                        ));
                    } catch (Exception e) {
                        log.error("recover failed!", e);
                    }
                }
                Container.getConsumeQueueConsumeHandler().ack(BrokerConstant.DELAY_TOPIC_NAME,
                        BrokerConstant.RECOVER_DELAY_MSG_CONSUMER_GROUP, otherDelayQueueId);
            });
        }
    }
}
